package org.yumeko.kafka.producer.groupoffset;

//需创建 tp_group_offset 主题
//./kafka-topics.sh --create --topic tp_group_offset --partitions 3 --replication-factor 1 --bootstrap-server 192.168.88.134:9092
//如果消费者停⽌，查看偏移量信息：
//./kafka-consumer-groups.sh --bootstrap-server 192.168.88.134:9092  --group group --describe
//将偏移量设置为最早的
//./kafka-consumer-groups.sh --bootstrap-server 192.168.88.134:9092 --group group --topic tp_group_offset --reset-offsets --to-earliest
//将偏移量设置为最新的
//./kafka-consumer-groups.sh --bootstrap-server 192.168.88.134:9092 --group group --topic tp_group_offset --reset-offsets --to-latest
//搜所有分区下的偏移量移量向前移动10个消息
//./kafka-consumer-groups.sh --bootstrap-server 192.168.88.134:9092 --group group --topic tp_group_offset --reset-offsets --shift-by -10
//分别将指定主题的指定分区的偏移量向前移动10个消息：
//./kafka-consumer-groups.sh --bootstrap-server 192.168.88.134:9092 --group group --topic tp_group_offset:1 --reset-offsets --shift-by -10
public class ProducerHandler implements Runnable {
    private String message;

    public ProducerHandler(String message) {
        this.message = message;
    }

    @Override
    public void run() {
        KafkaProducerSingleton kafkaProducerSingleton = KafkaProducerSingleton.getInstance();
        kafkaProducerSingleton.init("tp_group_offset", 3);
        int i = 0;

        while (true) {
            try {
                System.out.println("当前线程:" + Thread.currentThread().getName()
                        + "\t获取的kafka实例:" + kafkaProducerSingleton);
                kafkaProducerSingleton.sendKafkaMessage("发送消息: " + message + " " + (++i));
                Thread.sleep(100);
            } catch (Exception e) {
            }
        }
    }
}